package com.zzl.spark.recommend

import java.io.File

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd.RDD
import org.jfree.data.category.DefaultCategoryDataset
import org.joda.time.{DateTime, Duration}

object AlsEvaluation {

  def PrepareData():(RDD[Rating],RDD[Rating],RDD[Rating]) = {
    //1.创建用户评分数据
    val sc = new SparkContext(new SparkConf().setAppName("Recommend").setMaster("local[10]"))
    println("开始读取用户评分数据中...")
    val dataDir="file:/D:/ideawork/SparkLearning/"
    val rawUserData=sc.textFile(dataDir+"u.data")
    val rawRatings =rawUserData.map(_.split("\t").take(3))
    val ratingsRDD =rawRatings.map{
      case Array(user,movie,rating)=>Rating(user.toInt,movie.toInt,rating.toDouble)
    }
    println("共计:"+ratingsRDD.count().toString+" 条ratings")
    //2.创建电影ID与名称对照表
    println("开始读取电影数据中..")
    val itemRDD=sc.textFile(new File(dataDir,"u.item").toString)
    val movieTitle =itemRDD.map(line=>line.split("\\|").take(2)).map(array=>(array(0).toInt,array(1))).collect().toMap
    //3.显示数据记录数
    val numRationgs =ratingsRDD.count()
    val numUsers =ratingsRDD.map(_.user).distinct().count()
    val numMovies =ratingsRDD.map(_.product).distinct().count()
    println("共计: ratings: "+numRationgs+" User: "+numUsers+ " Movie: "+numMovies)
    //4.以随机的方式将数据分为3个部分
    val Array(trainData,validationData,testData) =ratingsRDD.randomSplit(Array(0.8,0.1,0.1))
    println("trainData:" +trainData.count()+" validationData:"+validationData.count()+" testData:"+testData.count())
    (trainData,validationData,testData)
  }

  def computeRmse(model: MatrixFactorizationModel, RatingRDD: RDD[Rating]):Double = {
    val num =RatingRDD.count()
    val perdictedRDD =model.predict(RatingRDD.map(r=>(r.user,r.product)))
    val rdd = RatingRDD.map(r=>((r.user,r.product),r.rating))
    val predictedAndRationgs=perdictedRDD.map(p=>((p.user,p.product),p.rating)).join(rdd).values
    math.sqrt(predictedAndRationgs.map(x=>(x._1-x._2)*(x._1-x._2)).reduce(_+_)/num)
  }


  /**
    * 输入下列trainModer 训练模型程序代码
    *
    * @param trainData
    * @param validationData
    * @param rank
    * @param numIterations
    * @param lambda
    */
  def trainModel(trainData: RDD[Rating], validationData: RDD[Rating], rank: Int, iterations: Int, lambda: Double):(Double,Double) = {
    val startTime = new DateTime()
    val model =ALS.train(trainData,rank,iterations,lambda)
    val endTime =new DateTime()
    val rmse =computeRmse(model,validationData)
    val duration = new Duration(startTime,endTime)
    println(f"训练参数:rank:$rank%3d,iterations:$iterations%.2f,lambda=$lambda 结果Rmse=$rmse%.2f"+" 训练时间需要"+duration.getMillis+"毫秒" )
    (rmse,duration.getStandardSeconds)
  }

  /**
    * 评估rank参数时  numIterations=Array(10) lambda=Array(10)
    * 评估numIterations时  rank=Array(10) lambda=Array(0.1)
    * 评估lambda时  rank=Array(10) numIterations=Array(10)
 *
    * @param trainData
    * @param validationData
    * @param str
    * @param rankArray
    * @param numIterationsArray
    * @param lambdaArray
    */
  def evaluateParameter(trainData: RDD[Rating], validationData: RDD[Rating], evaluateParameter: String, rankArray: Array[Int], numIterationsArray: Array[Int], lambdaArray: Array[Double]) = {
    var dataBarChart = new DefaultCategoryDataset()
    var dataLineChart = new DefaultCategoryDataset()
    for(rank<-rankArray;numIterations<-numIterationsArray;lambda<-lambdaArray){
      val (rmse,time) =trainModel(trainData,validationData,rank,numIterations,lambda)
      val parameterData = evaluateParameter match {
        case "rank"   => rank;
        case "numIterations" => numIterations;
        case "lambda" => lambda
      }
      dataBarChart.addValue(rmse,evaluateParameter,parameterData.toString)
      dataLineChart.addValue(time,"Time",parameterData.toString())
      Chart.plotBarLineChaet("ALS EVALUATIONS "+evaluateParameter,evaluateParameter,"RMSE",0.58,5,"Time",dataBarChart,dataLineChart)
    }
  }

  /**
    * 所有的参数交叉评估,,找出最好的参数组合并传入
    * 以上参数共有5*5*5=125个排列组合
    * @param trainData
    * @param validationData
    * @param rankArray
    * @param numIterationsArray
    * @param lambdaArray
    * @return
    */
  def evaluateAllParameter(trainData: RDD[Rating], validationData: RDD[Rating], rankArray: Array[Int], numIterationsArray: Array[Int], lambdaArray: Array[Double]):MatrixFactorizationModel = {
    val evaluations=for(rank<-rankArray;numIterations<-numIterationsArray;lambda<-lambdaArray)yield {
      val (rmse,time) =trainModel(trainData,validationData,rank,numIterations,lambda)
      (rank,numIterations,lambda,rmse)
    }
    val Eval =(evaluations.sortBy(_._4))
    val BestEval =Eval(0)
      println("最佳model参数:rank"+BestEval._1+"  ,iterations:"+BestEval._2+"  ,lambda:"+BestEval._3+" ,结果 rmse="+BestEval._4)
    val  bestModel =ALS.train(trainData,BestEval._1,BestEval._2,BestEval._3)
    bestModel
  }

  def trainValidation(trainData: RDD[Rating], validationData: RDD[Rating]):MatrixFactorizationModel = {
    println("-------------评估rank参数使用-----------------")
    evaluateParameter(trainData,validationData,"rank",Array(5,10,15,20,50,100),Array(10),Array(0.1))
    println("-------------评估numIterations-----------------")
    evaluateParameter(trainData,validationData,"numIterations",Array(10),Array(5,10,15,20,25),Array(0.1))
    println("-------------评估lamba-----------------")
    evaluateParameter(trainData,validationData,"lambda",Array(10),Array(10),Array(0.05,0.1,1,5,10.0))
    val bestModel=evaluateAllParameter(trainData,validationData,Array(5,10,15,20,25),Array(5,10,15,20,25),Array(0.05,0.1,1,5,10.0))
    bestModel
  }

  def main(args: Array[String]):Unit = {
    Recommend.SetLogger
    println("===========数据准备阶段============")
    val (trainData,validationData,testData) = PrepareData()
    trainData.persist()
    validationData.persist()
    testData.persist()
    println("===========训练阶段================")
    val bestModel =trainValidation(trainData,validationData)
    println("===========测试阶段================")
    val testRmse = computeRmse(bestModel,testData)
    println("使用testData测试bestModel, 结果rmse ="+testRmse)
    trainData.unpersist()
    validationData.unpersist()
    testData.unpersist()
  }
}
